%%%-------------------------------------------------------------------
%%% File    : svc_queue.erl
%%% Author  :  <vjache>
%%% Description : 	In-memory queue service that supports message 
%%%					priority and controlable execution concurency.
%%%					Message priority vary in range 0...9, 0 is
%%%					a highest priority and 9 is a lowest one. Default
%%%					concurency, i.e. the number of simultaneously 
%%%					executed jobs, equals to the number of schedulers 
%%%					in a system (see erlang:system_info(schedulers)). 
%%%
%%% Created : 23 Jun 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(svc_queue).

-behaviour(gen_server).

-include_lib("eunit/include/eunit.hrl").

-include("../include/util.hrl").

%% API
-export([start_link/1,
		 enqueue/3,
		 suspend/1,
		 suspend/2,
		 suspend/4,
		 resume/1,
		 resume/2,
		 info/1,
		 info_cluster/2,
		 test/2,
		 test/3,
		 test/1]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	 terminate/2,  code_change/3]).

-define(SERVER,?MODULE).
-define(CHECKERR(Expr,Format,Args), checkerr(Expr,Format,Args)).
%% intrinsic definitions
-record(state, {
				name=?SERVER,
				subqueue_by_prio=array:new(
								   [
									{size,10},
									{fixed,true},
									{default,
									 queue:new() 	%% awaiting tasks
									}]),
				postponed=queue:new(),
				postpone_timeout=10000,
				pool_size=erlang:system_info(schedulers),
				active=[],
				suspenders=dict:new(),
				leases=dict:new()
			   }).

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
%% Description: Starts the server
%%--------------------------------------------------------------------
start_link(QueueConfig) ->
    gen_server:start_link({local, element(1,QueueConfig)}, ?MODULE, [QueueConfig], []).

stop(QueueName) ->
	srv_call(QueueName,stop, stop_err).
%%--------------------------------------------------------------------
%% Function: 	enqueue(QueueName, Func, Priority) -> ok
%% Description: Puts message to queue with specified priority.
%%				QueueName - atom(). Actually this is a registered name
%%					of queue service.
%%				Func - fun() -> void() | {M,F,A}. Task to execute, 
%%					returned value ignored. If task function throws 
%%					{postpone,Reason} then task will be putted to the 
%%					postponed queue and reexecuted later.
%%				Priority - int() in range 0..9. Priority means that 
%%					messages with higher priority will be taken 
%%					earlier than with lower ones. 0 is a highest 
%%					priority, 9 is a lowes one.
%%--------------------------------------------------------------------
enqueue(QueueName, Func, Priority) when 
  is_atom(QueueName),
  is_function(Func, 0),
  is_integer(Priority),
  Priority>=0,
  Priority=<9 ->
	srv_call(QueueName, {enqueue,Func,Priority},enqueue_err);
enqueue(QueueName, {Module,Function,ArgList}=_MFA, Priority) when 
  is_atom(QueueName),
  is_atom(Module),
  is_atom(Function),
  is_list(ArgList),
  is_integer(Priority),
  Priority>=0,
  Priority=<9 ->
	srv_call(QueueName, {enqueue,
							  fun()->
									  apply(Module, Function, ArgList)
							  end,Priority},enqueue_err).
%%--------------------------------------------------------------------
%% Function: 	suspend(QueueName) -> ok
%% Description: Suspends entire queue. This mean that active (curently 
%%				executing) tasks will stil be executed but no further 
%%				messages awaiting in queue will be taken for execution 
%%				until resume(QueueName) will be called.
%%				QueueName - atom(). Actually this is a registered name
%%					of queue service.
%%--------------------------------------------------------------------
suspend(QueueName) ->
	suspend(QueueName,0).
%%--------------------------------------------------------------------
%% Function: 	suspend(QueueName,Priority) -> ok
%% Description: Suspends sub queues having lower or equal priorities 
%%				than specified one. This mean that active (curently 
%%				executing) tasks will stil be executed but no further 
%%				messages awaiting in restricted sub queues will be 
%%				taken for execution	until resume(QueueName,Priority) 
%%				will be called.
%%				QueueName - atom(). Actually this is a registered name
%%					of queue service.
%%--------------------------------------------------------------------
suspend(QueueName,Priority) ->
	srv_call(QueueName, {suspend, Priority},suspend_err).
suspend(QueueName,Priority,LeaseName,Timeout) when is_integer(Timeout) ->
	gen_server:cast(QueueName, {suspend, Priority, LeaseName, Timeout}).
%%--------------------------------------------------------------------
%% Function: 	resume(QueueName) -> ok
%% Description: Resumes entire queue. See suspend(QueueName).
%%--------------------------------------------------------------------
resume(QueueName) ->
	resume(QueueName,0).
%%--------------------------------------------------------------------
%% Function: 	resume(QueueName,Priority) -> ok
%% Description: Resumes sub queues having priorities lower that or 
%% 				equal specified one. See suspend(QueueName,Priority).
%%--------------------------------------------------------------------
resume(QueueName,Priority) ->
	srv_call(QueueName, {resume, Priority},resume_err).
%%--------------------------------------------------------------------
%% Function: 	info(QueueName) -> ok
%% Description: Returns summary information about queue state.
%%--------------------------------------------------------------------
info(QueueName) ->
	Bin=gen_server:call(QueueName, info,infinity),
	StateSnap=binary_to_term(Bin),
	info_report(StateSnap).

info_report(#state{
				   name=_Name,
				   subqueue_by_prio=Array,
				   pool_size=Concurency,
				   active=Active,
				   suspenders=Suspenders,
				   postponed=PQueue,
				   postpone_timeout=PTimeout,
				   leases=_Leases
				  }) ->
	PrioQueuesInfo=array:foldr(fun(Prio,Q,QIList)->
									   [{Prio,queue:len(Q)}|QIList]
							   end, [], Array),
	[
	 {active,length(Active),'of',Concurency},
	 {suspenders,dict:fetch_keys(Suspenders)},
	 {prio_queues,
	  PrioQueuesInfo
	 },
	 {postponed,queue:len(PQueue)},
	 {postpone_timeout,[PTimeout]}
	].

info_cluster(QueueName, merge) ->
	{Nodes,_}=svc_rrs:get_nodes(),
	{Replies,_BadNodes}=gen_server:multi_call(Nodes, QueueName, info, infinity),
	lists:foldl(fun
				   ({_Node,Bin},nil) ->
						info_report(binary_to_term(Bin));
				   ({_Node,Bin},AccIn) ->
						[{active,Active,'of',Concurency},
						 {suspenders,Suspenders},
						 {prio_queues,PrioQueuesInfo},
						 {postponed,PQueueLen},
	 					 {postpone_timeout,PTimeout}
						]=info_report(binary_to_term(Bin)),
						[{active,Active1,'of',Concurency1},
						 {suspenders,Suspenders1},
						 {prio_queues,PrioQueuesInfo1},
						 {postponed,PQueueLen1},
	 					 {postpone_timeout,PTimeout1}
						]=AccIn,
						[{active,Active+Active1,'of',Concurency+Concurency1},
						 {suspenders,Suspenders++Suspenders1},
						 {prio_queues,merge_prio_queues_info(PrioQueuesInfo, PrioQueuesInfo1)},
						 {postponed,PQueueLen+PQueueLen1},
	 					 {postpone_timeout,PTimeout++PTimeout1}
						]			
				end, nil, Replies).

merge_prio_queues_info(PrioQueuesInfo, PrioQueuesInfo1) ->
	lists:zipwith(fun({Prio,QLength},{Prio,QLength1})->
						  {Prio,QLength+QLength1}
				  end, PrioQueuesInfo, PrioQueuesInfo1).

%%====================================================================
%% gen_server callbacks
%%====================================================================

%%--------------------------------------------------------------------
%% Function: init(Args) -> {ok, State} |
%%                         {ok, State, Timeout} |
%%                         ignore               |
%%                         {stop, Reason}
%% Description: Initiates the server
%%--------------------------------------------------------------------
init([{QName,default}=_QueueConfig]) ->
	{ok, #state{name=QName}};
init([{QName,Concurency}=_QueueConfig]) when is_integer(Concurency)->
    {ok, #state{name=QName,pool_size=Concurency}}.

%%--------------------------------------------------------------------
%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, Reply, State} |
%%                                      {stop, Reason, State}
%% Description: Handling call messages
%%--------------------------------------------------------------------
handle_call(stop, _From, State) ->
	{stop, shutdown , ok, State};
handle_call(Request, From, State) ->
	try call({Request, From, State}) of
		{Reply, State1} ->
			{reply, Reply, State1}
	catch
		throw:Reason ->
			{reply, {error,{Reason,at,erlang:get_stacktrace()}}, State}
	end.

%%--------------------------------------------------------------------
%% Function: handle_cast(Msg, State) -> {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, State}
%% Description: Handling cast messages
%%--------------------------------------------------------------------
handle_cast(Msg, State) ->
	try cast({Msg, State}) of
		State1 ->
			{noreply, State1}
	catch
		throw:Reason ->
			{reply, {error,{Reason,at,erlang:get_stacktrace()}}, State}
	end.

%%--------------------------------------------------------------------
%% Function: handle_info(Info, State) -> {noreply, State} |
%%                                       {noreply, State, Timeout} |
%%                                       {stop, Reason, State}
%% Description: Handling all non call/cast messages
%%--------------------------------------------------------------------
handle_info(_Info, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% Function: terminate(Reason, State) -> void()
%% Description: This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any necessary
%% cleaning up. When it returns, the gen_server terminates with Reason.
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, _State) ->
    ok.

%%--------------------------------------------------------------------
%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
%% Description: Convert process state when code is changed
%%--------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
%% Handlers
call({poll_queue, _From, State}) ->
	case decide_msg(State) of
		busy ->
			{ok,State};
		Decision ->
			{ok,_State2=start_jobs(Decision)}
	end;
call({{enqueue,_Func,_Priority}=Msg, _From, State}) ->
	State1=put_to_subqueue(Msg,State),
	call({poll_queue, _From, State1});
call({enqueue_from_postponed=_Msg, From, State}) ->
	PQueue=State#state.postponed,
	case queue:out(PQueue) of
		{{value, Msg}, PQueue2} ->
			State1=State#state{postponed=PQueue2},
			call({Msg,From,State1});
		{empty, PQueue} ->
			throw({no_postponed_tasks})
	end;
call({{postpone,Func,Priority}=_Msg, _From, State}) ->
	PQueue=State#state.postponed,
	State1=State#state{postponed=queue:in({enqueue,Func,Priority}, PQueue)},
	{ok,Tref}=?CHECKERR(
			  timer:apply_after(
				State#state.postpone_timeout, 
				gen_server, call,[State#state.name,enqueue_from_postponed,infinity]),
			  "postpone failure",[]),
	{Tref,State1};
call({{job_finished,JobPid}=_Msg, _From, State}) ->
	JobList=State#state.active,
	State1=State#state{active=lists:delete(JobPid,JobList)},
	call({poll_queue, _From, State1});
call({{suspend,Priority}, {SuspenderPid,_Tag}=_From, State}) ->
	SuspenderId={SuspenderPid,Priority},
	Suspenders=State#state.suspenders,
	case dict:find(SuspenderId, Suspenders) of
		{ok,_MonitorPid} ->
			Suspenders1=dict:update(SuspenderId, 
									fun({SuspCnt,MonPid}) ->
											{SuspCnt+1,MonPid}
									end, Suspenders),
			{ok,State#state{suspenders=Suspenders1}};
		error ->
			SrvPid=self(),
			Suspenders1=dict:store(
						  SuspenderId,
						  {1,spawn_link(
							 fun() ->
									 resumer_init(State,SuspenderId,SrvPid)
							 end)}, Suspenders),
			{ok,State#state{suspenders=Suspenders1}}
	end;
call({{resume_all,SuspenderId}, From, State}) when is_tuple(SuspenderId) ->
	Suspenders=State#state.suspenders,
	case dict:find(SuspenderId, Suspenders) of
		error ->
			{ok,State};
		{ok,{_,MonitorPid}} ->
			State1=remove_suspenders_and_notify(
					 State,SuspenderId,MonitorPid),
			call({poll_queue, From, State1})
	end;
call({{resume,SuspenderId}, From, State}) when is_tuple(SuspenderId) ->
	Suspenders=State#state.suspenders,
	case dict:find(SuspenderId, Suspenders) of
		error ->
			throw({resume_error,{SuspenderId}});
		{ok,{1,MonitorPid}} ->
			State1=remove_suspenders_and_notify(
					 State,SuspenderId,MonitorPid),
			call({poll_queue, From, State1});
		{ok,{SuspCnt,MonitorPid}} ->
			Suspenders1=dict:update(
						  SuspenderId, 
						  fun(_) ->
								  {SuspCnt-1,MonitorPid}
						  end,Suspenders),
			State1=State#state{suspenders=Suspenders1},
			call({poll_queue, From, State1})
	end;
%% call({{resume,SuspenderId}, _From, State}) when is_tuple(SuspenderId) ->
%% 	Suspenders=State#state.suspenders,
%% 	Suspenders1=dict:erase(SuspenderId, Suspenders),
%% 	case Suspenders of
%% 		Suspenders1 ->
%% 			throw({resume_error,{SuspenderId}});
%% 		_ ->
%% 			ok
%% 	end,
%% 	State1=State#state{suspenders=Suspenders1},
%% 	call({poll_queue, _From, State1});
call({{resume,Priority}, {SuspenderPid,_Tag}=From, State}) when is_integer(Priority) ->
	call({{resume,_SuspenderId={SuspenderPid,Priority}}, From, State});
call({info, _From, State}) ->
	{term_to_binary(State,[{compressed,1}]),State}.

cast({{suspend, Priority, LeaseName, Timeout}=_Msg, State}) ->
	Leases=State#state.leases,
	LeaserId={LeaseName,Priority},
	case dict:find(LeaserId, Leases) of
		error ->
			Pid=start_leaser(State#state.name,infinity,Priority),
			Leases1=dict:store(LeaserId, Pid, Leases),
			Pid ! {update_lease,Timeout},
			State#state{leases=Leases1};
		{ok,Pid} ->
			Pid ! {update_lease,Timeout},
			State
	end.
start_leaser(QName,Timeout,Priority) ->
	spawn_link(
	  fun()->
			  leaser_loop(QName,Timeout,Priority)
	  end).

leaser_loop(QName,Timeout,Priority) ->
	receive
		{update_lease,Timeout1} when Timeout==infinity ->
			suspend(QName,Priority),
			leaser_loop(QName,Timeout1,Priority);
		{update_lease,Timeout1} ->
			leaser_loop(QName,Timeout1,Priority)
		after 
			Timeout ->
			resume(QName,Priority),
			leaser_loop(QName,infinity,Priority)
	end.
resumer_init(State,SuspenderId={SuspenderPid,_Priority},SrvPid) ->
	MonRef=erlang:monitor(process, SuspenderPid),
	resumer_loop(State,SuspenderId,SrvPid,MonRef).
resumer_loop(State,SuspenderId={SuspenderPid,_Prio},SrvPid,MonRef) ->
	receive
		{'DOWN', MonRef, process, SuspenderPid, _Info} ->
			resume_all(State,SuspenderId),
			resumer_loop(State,SuspenderId,SrvPid,MonRef);
		{resumed,SrvPid} ->
			ok
	end.

remove_suspenders_and_notify(State,SuspenderId,MonitorPid) ->
	Suspenders=State#state.suspenders,
	MonitorPid ! {resumed,self()},
	Suspenders1=dict:erase(SuspenderId, Suspenders),
	State#state{suspenders=Suspenders1}.

put_to_subqueue({enqueue,_Func,Priority}=Msg,State) ->
	Array=State#state.subqueue_by_prio,
	Q=array:get(Priority, Array),
	State#state{subqueue_by_prio=array:set(Priority, queue:in(Msg, Q), Array)}.

decide_msg(State) ->
	Suspenders=State#state.suspenders,
	SuspCount=dict:size(Suspenders),
	Array=State#state.subqueue_by_prio,
	JobList=State#state.active,
	ActiveCount=length(JobList),
	Quota=State#state.pool_size-ActiveCount,
	if 
		Quota>0 ->
			QList0=array:to_list(Array),
			QList=case SuspCount of
					  0 ->
						  QList0;
					  _ ->
						  PrioSuspend=find_suspender_highest_priority(Suspenders),
						  lists:sublist(QList0, PrioSuspend)
				  end,
			case seek_queue(QList,Quota) of
				false ->
					busy;
				{Queue,MsgList,Prio} ->
					Array1=array:set(Prio, Queue, Array),
					{State#state{subqueue_by_prio=Array1}, {Prio,MsgList}}
			end;
		true ->
			busy
	end.

find_suspender_highest_priority(Suspenders) ->
	dict:fold(fun
				 ({_Pid,Priority}=_SuspenderId,_,AccIn) when Priority<AccIn->
					  Priority;
				 (_SuspenderId,_,AccIn)->
					  AccIn
			  end, 10, Suspenders).
	
seek_queue(QueueList,Max) ->
	seek_queue(QueueList,Max,0).
seek_queue([]=_QueueList,_Max,_Prio)->
	false;
seek_queue([Queue|Tail]=_QueueList,Max,Prio)->
	case out_n(Queue,Max) of
		{Queue,[]} ->
			seek_queue(Tail,Max,Prio+1);
		{Queue2,Items} ->
			{Queue2,Items,Prio}
	end.

out_n(Queue,Max) ->
	out_n(Queue,Max,[]).
out_n(Queue,0,Items) ->
	{Queue,Items};
out_n(Queue,N,Items) ->
	case queue:out(Queue) of
		{{value, Item}, Queue2}->
			out_n(Queue2,N-1,[Item|Items]);
		{empty, Queue} ->
			{Queue,Items}
	end.

start_jobs({State,{_Priority,MsgList}}=_Decision) ->
%% 	io:format("~p Start ~p ~p~n",[now(),Priority,length(MsgList)]),
	JobPids=[spawn_link(
			   fun()->
					   job_init(State,Msg)
			   end
			   )||Msg<-MsgList],
	JobList=State#state.active,
	State#state{active=JobList++JobPids}.

job_init(State,{enqueue,Func,Priority}=_Msg) ->
	try Func() of
		_Result ->
			?LOG_TRACE("Job finished. \nPriority=~p, \nResult=~p",[Priority,_Result])
	catch
		throw:{postpone,Reason} ->
			?REP_WARN([{queue,State#state.name},
					   {job_postponed,Reason},
					   {priority,Priority},
					   {at,erlang:get_stacktrace()}]),
			job_postpone(State,{enqueue,Func,Priority});
		_:Reason ->
			?REP_ERR([
					  {queue,State#state.name},
					  {priority,Priority},
					  {error,Reason},
					  {at,erlang:get_stacktrace()}
					 ])
	after
		job_release(State,self())
	end.

job_postpone(State,{enqueue,Func,Priority}) ->
	gen_server:call(State#state.name, 
					{postpone,Func,Priority},infinity).

job_release(State,JobPid) ->
	gen_server:call(State#state.name, {job_finished,JobPid}, infinity).

%% resume_(State,SuspenderId) when 
%%   is_record(State, state), 
%%   is_tuple(SuspenderId) ->
%% 	gen_server:call(State#state.name, {resume,SuspenderId},infinity).
resume_all(State,SuspenderId) when 
  is_record(State, state), 
  is_tuple(SuspenderId) ->
	srv_call(State#state.name, {resume_all,SuspenderId},resume_all_err).
srv_call(PoolName,Request,ErrTag) ->
	checkerr(gen_server:call(PoolName, Request, infinity),ErrTag).
%% Helper Utilities
checkerr({aborted,ErrReason},Format,Args) ->
	checkerr({error,ErrReason},Format,Args);
checkerr({error,ErrReason},Format,Args) ->
	throw({queue_error,ErrReason,lists:flatten(io_lib:format(Format, Args))});
checkerr(Result,_Format,_Args) ->
	Result.

checkerr({error,Reason},ErrTag) ->
	throw({ErrTag,Reason});
checkerr(Result,_ErrTag) ->
	Result.

%%--------------------------------------------------------------------
%%% Test functions
%%--------------------------------------------------------------------
test(seek,Max) ->
	QList=[queue:from_list([m01,m02,m03,m04]),
	queue:from_list([m11,m12,m13]),
	queue:from_list([m21,m22,m23,m24,m25])],
	seek_queue(QList,Max);
test(seek1,Max) ->
	QList=[queue:from_list([]),
	queue:from_list([m11,m12,m13]),
	queue:from_list([m21,m22,m23,m24,m25])],
	seek_queue(QList,Max).
test(seek,Prio,Max) ->
	QList=[queue:from_list([]),
	queue:from_list([m11,m12,m13]),
	queue:from_list([m21,m22,m23,m24,m25])],
	seek_queue([lists:nth(Prio+1, QList)],Max,Prio).

test(enqueue) ->
	[enqueue(def_queue,fun()->
					receive
					after 1000 ->
							io:format("Job finished. Prio ~p. ~n",[Prio])
					end
			end,Prio)||Prio<-[0,0,0,0,0,0,1,1,1,1,1,1,2,2,2,3,3,4,5,5,5,5,5,6,7,8]];

	
test(enqueue1) ->
	[
	 begin
		 Prio=random:uniform(10)-1,
		 enqueue(def_queue,fun()->
						 receive
							 after 1000 ->
								 io:format("Job finished. Prio ~p. ~n",[Prio])
						 end
				 end,Prio)
	 end||_<-lists:seq(1, 100)];

test(Priority) when is_integer(Priority) ->
	[enqueue(def_queue,fun()->
					receive
					after 1000 ->
							io:format("Job finished. Prio ~p. ~n",[Prio])
					end
			end,Prio)||Prio<-[Priority,Priority,Priority,Priority,Priority]].

%%%---------------------------------------------------------------------
%%% eunit
%%%---------------------------------------------------------------------

basic_test_() ->
	{foreach, 
	 local, 
	 fun setup_/0, 
	 fun teardown_/1, 
	 [fun(QueueName)->
			  ?_test(test_suspend_resume(QueueName))
	  end,
	  fun(QueueName)->
			  ?_test(test_2suspend_2resume(QueueName))
	  end,
	  fun(QueueName)->
			  ?_assertThrow({resume_err,_},test_2suspend_3resume(QueueName))
	  end,
	  fun(QueueName)->
			  ?_test(test_suspend_lease(QueueName))
	  end,
	  fun(QueueName)->
			  ?_test(test_suspend_2lease(QueueName))
	  end,
	  fun(QueueName)->
			  ?_test(test_suspend_lease_prolongate(QueueName))
	  end,
	  fun(QueueName)->
			  ?_test(test_enqueue(QueueName))
	  end
	 ]}.

setup_() ->
	ReplyPid=self(),
	QueueName=test_queue,
	spawn(fun()->
				  ReplyPid ! start_link({QueueName,default})
		  end),
	receive
		M ->
			{ok,_Pid}=M
	end,
	QueueName.
teardown_(QueueName) ->
	stop(QueueName).


test_suspend_resume(QueueName) ->
	suspend(QueueName),
	resume(QueueName).
test_2suspend_2resume(QueueName) ->
	suspend(QueueName),
	suspend(QueueName),
	resume(QueueName),
	resume(QueueName).
test_2suspend_3resume(QueueName) ->
	suspend(QueueName),
	suspend(QueueName),
	resume(QueueName),
	resume(QueueName),
	resume(QueueName).
test_suspend_lease(QueueName) ->
	suspend(QueueName,0,test_lease,200),
	timer:sleep(30),
	?assertMatch({value,{suspenders,[_]}},
				 lists:keysearch(suspenders, 1, info(QueueName))),
	timer:sleep(400),
	?assertEqual({value,{suspenders,[]}},
				 lists:keysearch(suspenders, 1, info(QueueName))).
test_suspend_2lease(QueueName) ->
	suspend(QueueName,0,test_lease,200),
	suspend(QueueName,0,test_lease1,200),
	timer:sleep(30),
	?assertMatch({value,{suspenders,[_,_]}},
				 lists:keysearch(suspenders, 1, info(QueueName))),
	timer:sleep(400),
	?assertEqual({value,{suspenders,[]}},
				 lists:keysearch(suspenders, 1, info(QueueName))).
test_suspend_lease_prolongate(QueueName) ->
	suspend(QueueName,0,test_lease,200),
	timer:sleep(30),
	?assertMatch({value,{suspenders,[_]}},
				 lists:keysearch(suspenders, 1, info(QueueName))),
	timer:sleep(100),
	suspend(QueueName,0,test_lease,200),
	timer:sleep(30),
	?assertMatch({value,{suspenders,[_]}},
				 lists:keysearch(suspenders, 1, info(QueueName))),
	timer:sleep(400),
	?assertEqual({value,{suspenders,[]}},
				 lists:keysearch(suspenders, 1, info(QueueName))).

test_enqueue(QueueName) ->
	ReplyPid=self(),
	Msg=make_ref(),
	enqueue(QueueName, 
			fun() ->
					ReplyPid ! Msg
			end, 
			1),
	receive
		Msg ->
			ok
		after 
			200 ->
			throw(job_not_executed)
	end.
